package AdDemo

import java.text.SimpleDateFormat
import java.util.Date

import Sun.BasicFunction
import breeze.linalg.diff
import ch.epfl.lamp.compiler.msil.emit.Label
import org.apache.spark.mllib.linalg.{Vector, Vectors}
import org.apache.spark.mllib.optimization.Gradient
import org.apache.spark.mllib.util.MLUtils

import scala.collection.immutable.List
import scala.collection.mutable.ArrayBuffer
import org.apache.spark.annotation.DeveloperApi
import org.apache.spark.mllib.linalg.{DenseVector, Vector, Vectors}
//import org.apache.spark.mllib.linalg.BLAS.{axpy, dot, scal}


/**
 * Created by SHANGMAI on 2016/6/6.
 */

class MyGradient(numClasses: Int) extends Gradient {

  def this() = this(2)


  //返回向量与损失函数
  override def compute(data: Vector, label: Double, weights: Vector): (Vector, Double) = {

    val (gradient,loss) = gradientAndLoss(data, label, weights)
    return (gradient,loss)
  }




  override def compute(
                        data: Vector,
                        label: Double,
                        weights: Vector,
                        cumGradient: Vector): Double = {


    val (gradient,loss) = gradientAndLoss(data, label, weights)

    //进行简单的覆盖
    val cumValues = cumGradient.toDense.values


    for(i <- 0 to cumValues.size - 1)
      cumValues(i) = gradient.toDense.values(i)

    var num = 0
    for(i <- 0 to weights.size - 1)
      if(weights.apply(i) != 1)
        num = num + 1


    //尝试给出效果
    //axpy(loss, data, cumGradient)
    //umGradient= cumGradient.apply(1) + data.apply(1)   c
    //进行向量的相加，导出新的向量
    return loss
  }


  //算单个数据的损失 极大似然
  //整个长度为    2m+2n+11
  //准备返回LOSS和梯度
  def gradientAndLoss(data: Vector, label: Double, weights: Vector): (Vector, Double) = {
    //vector按channel和时间分特征
    //处理之后向量 分成这几个部分
    //第一是用户  第二是C feature  之后就是channel id的聚合的了
    var dataResult: Map[(Double, Double), Vector] = Map()
    var tempList = ArrayBuffer[Double]()
    var key = (1.0, 1.0)

    //这里设置起始时间,观察转换时间是否为空当其没有转换的时候选择现在时间
    if(label == 0 || data.apply(0) == "null")
      BasicFunction.startTime = {
      val sdf = new SimpleDateFormat("yyyy-MM-dd");
      val nowtime = sdf.parse("2013-07-10").getTime
      nowtime
    }
    //如果有转换时间取转换时间
    else
      BasicFunction.startTime = data.apply(0).toLong

    //进行数据的结构梳理，由于前面增加了偏移值，所以这里多偏移一次（appendBias）
    for (i <- BasicFunction.info_length + BasicFunction.featureC_length to data.size - 2) {


      //channel开始计算,并整理新的channel,
      if ((i - BasicFunction.info_length - BasicFunction.featureC_length) % (BasicFunction.featureD_length + BasicFunction.featureE_length + 2) == 0) {
        //不是第一个情况下，或者最后一个情况下
        if (i != BasicFunction.info_length + BasicFunction.featureC_length )
          dataResult += (key -> Vectors.dense(tempList.toArray))

        //把channel和time字段填充
        key = (data.apply(i), data.apply(i + 1))
        tempList = ArrayBuffer[Double]()
      }

      //将数据填充到结构里,非时间字段填充,去掉channel和time字段字段
      else if ((i - BasicFunction.info_length - BasicFunction.featureC_length) % (BasicFunction.featureD_length + BasicFunction.featureE_length + 2) != 1)
        tempList += data.apply(i)

      //dataResult
    }

    //最后也应该添加
    dataResult += (key -> Vectors.dense(tempList.toArray))

    //给出H与S的值，这个值针对一条记录是不变的
    //这里的nowtime取到2013年7月1日

    val hazardRateValue = BasicFunction.hazardRateFunction(dataResult, weights)
    val survivalValue = BasicFunction.survivalFunction(dataResult, weights)

    //featureC的值
    val logisticValue = BasicFunction.logisticFuntion(data,weights)


    //debug
    if( hazardRateValue == 0 || survivalValue == 0 || logisticValue == 1 || logisticValue == 0 )
      {
        val error = 1
      }

    var loss = 0.0

    //当为转化的时候
    if (label == 1)
      loss = math.log(hazardRateValue) + math.log(survivalValue) + math.log(logisticValue)

    //当未转化的时候
    else
      loss = math.log((1 - logisticValue) * (1 - survivalValue))


    //debug，去掉重复点
    if( loss >  99999999 || loss < -999999999  || loss.isNaN)
    {
      loss = 0
    }


    //之后计算 给出梯度

    //分批计算梯度 再合并

    val featureC = BasicFunction.getStaticFeatureC(data)

    //梯度要和向量一致，多少维向量多少维地图
    var gradient = new Array[Double](weights.size)


    var gradientE:Vector = null
    var gradientD:Vector = null

    //使用dataResult 进行计算 得到最终的梯度结果
    //进行data的操作 按条读取
    //此处采取随机梯度下降的办法来进行编程(单条记录的梯度）
    //在此主要采集e和d的向量
    dataResult.keys.foreach { key =>
      //判断channel向量 {
      //获取值向量,获取所有的feature
      val x = dataResult(key)
      //获得渠道
      val channel = key._1

      //获得时间
      val time = key._2

      //进行tiem差值的计算,调小量纲
      val timediff = (BasicFunction.startTime - time) / BasicFunction.timedivision

      //测试使用，当timediff不小于0 的时候 进行操作
      if (timediff > 0) {
        //定义变量的featurE weightE featureD weightD
        val featureE = BasicFunction.getActionfeatureE(x)
        val weightE = BasicFunction.getActionWeightE(weights, channel)
        val featureD = BasicFunction.getTimeFeatureD(x)
        val weightD = BasicFunction.getTimeWeightD(weights, channel)

        //每次给出新的梯度 更新E

        gradientE = {
          var resultGradient: Vector = null
          //此处定义需要计算的向量
          val derivativeE = BasicFunction.derivativealpha(featureE, weightE)

          //在转换的时候，给出修改该渠道的向量
          if (label == 1) {
            val multip = 1 / hazardRateValue * BasicFunction.lambdaFuntion1(featureD, weightD, timediff) + BasicFunction.lambdaFuntion2(featureD, weightD, timediff)
            resultGradient = BasicFunction.multipVector(derivativeE, multip)
          }
          //非转换的时候也给出相应的向量
          else {
            val term1 = BasicFunction.lambdaFuntion2(featureE, weightE, timediff)
            val term2 = logisticValue * survivalValue / ((1 - logisticValue) * (1 - survivalValue)) * BasicFunction.lambdaFuntion2(featureD, weightD, timediff)
            resultGradient = BasicFunction.multipVector(derivativeE, term1 - term2)
          }

          resultGradient
        }
        gradient = alterVecor(channel, gradientE, gradient, "E")


        //每次给出新的梯度 更新D
        gradientD = {

          var resultGradient: Vector = null

          //设置共享项

          val shareterm = (1 - BasicFunction.lambdaFuntion2(featureD, weightD, timediff)) * timediff
          if (label == 1) {
            val term1 = (1 / hazardRateValue) * BasicFunction.alphaFuntion(BasicFunction.getActionfeatureE(x), BasicFunction.getActionWeightE(weights, channel))
            val coefficient1 = (1 - BasicFunction.lambdaFuntion2(featureD, weightD, timediff)) * (1 - BasicFunction.alphaFuntion(featureD, weightD) * timediff)
            val multip = term1 * coefficient1 + shareterm
            resultGradient = BasicFunction.multipVector(BasicFunction.derivativealpha(BasicFunction.getTimeFeatureD(x), BasicFunction.getTimeWeightD(weights, channel)), multip)
          }
          else {
            val multip = (1 - (logisticValue * survivalValue / ((1 - logisticValue) * (1 - survivalValue)))) * shareterm
            resultGradient = BasicFunction.multipVector(BasicFunction.derivativealpha(BasicFunction.getTimeFeatureD(x), BasicFunction.getTimeWeightD(weights, channel)), multip)
          }

          resultGradient
        }
        gradient = alterVecor(channel, gradientD, gradient, "D")
      }
    }




    //最后统一更新有关于C的梯度 derivative
    var gradientC:Vector = null

     //在有转化的情况下
    if(label == 1) {
      val multip = 1 / logisticValue
      gradientC = BasicFunction.multipVector(BasicFunction.derivativeC(data,weights), multip)
    }

    //没有的情况下
    else {
      val multip = (-1) * (1 - survivalValue) / (1 - logisticValue) * (1 - survivalValue)
      gradientC =  BasicFunction.multipVector(BasicFunction.derivativeC(data,weights), multip)
    }

    gradient = alterVecor(0,gradientC,gradient,"C")

    val resultGradient = Vectors.dense(gradient).toSparse


    return (resultGradient, loss)

  }

  //给出一个计算e和d的算法，来进行操作，用v的变量来修改original的值
  //original为传输的总向量，这里主要用于修改
  //vectortype主要为向量形式 c e d,根据不同的起始阶段修改不同的值
  def alterVecor(channel:Double,v:Vector,original:Array[Double],vectortype:String): Array[Double] =
  {
    var start = 0
    //分类型找到向量开始的位置
    val channelindex =  channel.toLong.toString
    if (vectortype == "C")
      start = BasicFunction.info_length

    //根据channel给出E的开始位置
    else if(vectortype == "E")
      start =  BasicFunction.indexChannel(channelindex)

    //根据channel给出D的出事位置
    else if(vectortype == "D")
      start =  BasicFunction.indexChannel(channelindex) + BasicFunction.featureE_length

    for (i <- start to start + v.size - 1){
      original(i) =  original(i) + v.apply(i - start)
    }
    return original
  }

}